package com.xinqing.bigdata.flink.util;

import com.alibaba.fastjson.JSON;
import com.xinqing.bigdata.flink.datastream.model.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.Random;
import java.util.UUID;

/**
 * @Author:CHQ
 * @Date:2021/5/12 15:51
 * @Description
 */
public class KafkaMsgProduct {
    public static void main(String args[]) {
        Logger logger = LoggerFactory.getLogger(KafkaMsgProduct.class);
        KafkaProducer<String, String> producer = null;

        try {
            Properties properties = new Properties();
            //broker
            properties.setProperty("bootstrap.servers", "10.201.7.175:9092");
            // key 序列化
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            // value 序列化
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

            //创建生产者
            producer = new KafkaProducer<>(properties);

            int count = 1;
            //每隔3s发送一条记录
            while (count <= 50000) {
                //String data = JSON.toJSONString(new UserCostInfo("1", new Random().nextDouble(), System.currentTimeMillis()));
                String data = JSON.toJSONString(new OrderInfo(
                        "1",
                        String.valueOf(new Random().nextInt(100)),
                        "2",
                        System.currentTimeMillis()
                ));
                //    String data = JSON.toJSONString(new UserClickAdInfo("1", "s", System.currentTimeMillis()));
                System.out.println(data);
                producer.send(new ProducerRecord<>("cost19", data));
                Thread.sleep(1000);
                count++;
            }
        } catch (Exception e) {
            logger.error("kafka product msg error ", e);
        } finally {
            if (producer != null) {
                //及时关闭资源
                producer.close();
            }
        }
    }
}
